package com.xzx.flink.streamapi.window;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.HashSet;

/**
 * 会用 PV/UV 这个比值，来表示“人均重复访问量”，也就是平均每个用户会访问多少次页面
 *
 * @author xinzhixuan
 * @version V1.0.0
 * @date 2021/4/20
 **/
public class Window_02_EventTime_AggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                    @Override
                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                        return element.getTimestamp();
                    }
                }))
                .keyBy(x -> true)
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(2)))
                .aggregate(new AggregateFunction<ClickEvent, Tuple2<HashSet<String>, Long>, Double>() {
                    @Override
                    public Tuple2<HashSet<String>, Long> createAccumulator() {
                        // 每个实例只会创建一次
                        return Tuple2.of(new HashSet<>(), 0L);
                    }

                    @Override
                    public Tuple2<HashSet<String>, Long> add(ClickEvent value, Tuple2<HashSet<String>, Long> accumulator) {
                        accumulator.f0.add(value.getUser());
                        accumulator.f1 = accumulator.f1 + 1;
                        return accumulator;
                    }

                    @Override
                    public Double getResult(Tuple2<HashSet<String>, Long> accumulator) {
                        return accumulator.f1 * 0.1 / accumulator.f0.size();
                    }

                    @Override
                    public Tuple2<HashSet<String>, Long> merge(Tuple2<HashSet<String>, Long> a, Tuple2<HashSet<String>, Long> b) {
                        //因为只有一个分区了，这里的merge不会被触发
                        return null;
                    }
                })
                .print();

        env.execute(Window_02_EventTime_AggregateFunction.class.getSimpleName());
    }
}
